package com.chezhibao.flink;

import com.virtusai.clickhouseclient.ClickHouseClient;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.regex.Pattern;

/**
 * 描述
 *
 * @author zhulixin
 */
public class Main {

    // Initialize client (endpoint, username, password)
    private static ClickHouseClient client = new ClickHouseClient("http://172.16.12.129:8123", "default", "123456");
    public static void main(String[] args) throws Exception {
        File file = new File("D:\\tmp\\temp\\dump\\clotho\\log");

        Pattern pattern = Pattern.compile("\\|");

        Set<String> set = new HashSet<>();
        List<Object[]> rows = new ArrayList<>();
        for (File f : file.listFiles()) {
            System.out.println("executing file: " + f);
            BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(f)));
            String line;
            while ((line = reader.readLine()) != null) {
                Object[] row = new Object[15];
                String[] str = pattern.split(line);
                String traceId = str[0];
                long startTime = Long.valueOf(str[1]);
                int rpcType = Integer.valueOf(str[2]);
                String appName = str[3];
                String rpcId = str[4];
                int span0 = 0;
                int span1;
                String traceName = "";
                String serviceName = "";
                String methodName = "";
                String remoteIp = "";
                String resultCode = "";
                int requestSize = 0;
                int responseSize = 0;
                String ext = "";


                if (rpcType == 0) {
                    span1 = Integer.valueOf(str[5]);
                    traceName = str[6];
                    resultCode = str[7];
                    if (str.length == 9) {
                        ext = str[8];
                    }
                } else if (rpcType == 1 || rpcType == 3 || rpcType == 4 || rpcType == 5) {
                    serviceName = str[5];
                    methodName = str[6];
                    remoteIp = str[7];
                    String span = str[8];
                    String[] spans = span.substring(1, span.length() - 1).split(",");
                    span0 = Integer.valueOf(spans[0].trim());
                    span1 = Integer.valueOf(spans[1].trim());
                    resultCode = str[9];
                    requestSize = Integer.valueOf(str[10]);
                    responseSize = Integer.valueOf(str[11]);
                    if (str.length == 13) {
                        ext = str[12];
                    }
                } else if (rpcType == 2) {
                    serviceName = str[5];
                    methodName = str[6];
                    remoteIp = str[7];
                    span1 = Integer.valueOf(str[8]);
                    resultCode = str[9];
                    requestSize = Integer.valueOf(str[10]);
                    responseSize = Integer.valueOf(str[11]);
                    if (str.length == 13) {
                        ext = str[12];
                    }
                } else if (rpcType == 6) {
                     serviceName = str[5];
                     methodName = str[6];
                     remoteIp = str[7];
                     if ("MQSend".equals(serviceName)) {
                         String span = str[8];
                         String[] spans = span.substring(1, span.length() - 1).split(",");
                         span0 = Integer.valueOf(spans[0].trim());
                         span1 = Integer.valueOf(spans[1].trim());
                     } else {
                         span1 = Integer.valueOf(str[8]);
                     }
                    resultCode = str[9];
                    requestSize = Integer.valueOf(str[10]);
                    responseSize = Integer.valueOf(str[11]);
                    if (str.length == 13) {
                        ext = str[12];
                    }
                } else if (rpcType == 9) {
                    continue;
                } else {
                    System.out.println("not process: " + line);
                    continue;
                }

                row[0] = traceId;
                row[1] = startTime;
                row[2] = rpcType;
                row[3] = appName;
                row[4] = rpcId;
                row[5] = traceName;
                row[6] = serviceName;
                row[7] = methodName;
                row[8] = remoteIp;
                row[9] = span0;
                row[10] = span1;
                row[11] = resultCode;
                row[12] = requestSize;
                row[13] = responseSize;
                row[14] = ext;
                rows.add(row);
                if (rows.size() == 500000) {
                    long start = System.currentTimeMillis();
                    insert(rows);
                    System.out.println("insert completed, cost: " + (System.currentTimeMillis() - start) + ", size: " + rows.size());
                    rows = new ArrayList<>();
                }
            }
            reader.close();
        }
        if (rows.size() > 0) {
            long start = System.currentTimeMillis();
            insert(rows);
            System.out.println("insert completed, cost: " + (System.currentTimeMillis() - start) + ", size: " + rows.size());
            Thread.sleep(2000);
        }
        client.close();
    }


    private static void insert(List<Object[]> rows) {
        // Insert data
        client.post("INSERT INTO default.trace", rows);
        // Close client
//        client.close();
    }
}
